package com.atbeijing.bigdata.spark.core.req

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark01_Req_HotCategoryTop10_3 {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
        val sc = new SparkContext(conf)

        val action = sc.textFile("data/user_visit_action.txt")

        // 创建累加器
        val acc = new HotCategoryAccumulator
        // 注册累加器
        sc.register(acc, "HotCategory")

        // 累加器的执行代码一般用在行动算子中，而不是转换算子
        action.foreach(
            line => {
                val datas = line.split("_")
                if ( datas(6) != "-1" ) {
                    // 点击的场合
                    acc.add((datas(6), "click"))
                } else if ( datas(8) != "null" ) {
                    // 下单的场合
                    val ids = datas(8).split(",")
                    ids.foreach(
                        id => {
                            acc.add((id, "order"))
                        }
                    )
                } else if (datas(10) != "null") {
                    // 支付的场合
                    val ids = datas(10).split(",")
                    ids.foreach(
                        id => {
                            acc.add((id, "pay"))
                        }
                    )
                }
            }
        )

        // 获取累加器的值
        val counts: List[HotCategoryCount] = acc.value.map(_._2).toList

        val top10 = counts.sortWith(
            (left, right) => {
                if ( left.clickCnt > right.clickCnt ) {
                    //返回true表示判断条件符合自己的排序意愿
                    true
                } else if ( left.clickCnt == right.clickCnt ) {
                    if ( left.orderCnt > right.orderCnt ) {
                        true
                    } else if ( left.orderCnt == right.orderCnt ) {
                        left.payCnt > right.payCnt
                    } else {
                        false
                    }
                } else {
                    false
                }
            }
        ).take(10)

        top10.foreach(println)

        sc.stop()

    }

    // 样例类中的参数可以被自动当成类的属性，默认参数值不能改，如果想要改，需要明确声明为var
    case class HotCategoryCount(cid:String, var clickCnt:Long, var orderCnt:Long, var payCnt:Long)
    /**
     * 热门品类累加器
     * 1. 继承AccumulatorV2类
     * 2. 定义泛型
     *     IN : 向累加器中输入的数据类型 : (品类，行为类型)
     *     OUT : 累加器的累加结果的类型 : Map[品类，HotCategoryCount]
     *                                Map[品类，(Int, Int, Int))]
     * 3. 重写方法
     */
    class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]] {

        private val hcmap = mutable.Map[String, HotCategoryCount]()

        override def isZero: Boolean = hcmap.isEmpty

        override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]] =  {
            new HotCategoryAccumulator()
        }

        override def reset(): Unit = hcmap.clear()

        override def add(v: (String, String)): Unit = {

            val ( cid, actiontype ) = v

            //要么获取到旧的,要么给个默认值
            val hc = hcmap.getOrElse(cid, HotCategoryCount(cid, 0L, 0L, 0L))

            actiontype match {
                case "click" => hc.clickCnt += 1
                case "order" => hc.orderCnt += 1
                case "pay" => hc.payCnt += 1
            }

            hcmap.update(cid, hc)
        }

        override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]]): Unit =  {
            val map = other.value
            map.foreach {
                case ( cid, otherHC ) => {
                    val thishc = hcmap.getOrElse(cid, HotCategoryCount(cid, 0L, 0L, 0L))

                    thishc.clickCnt += otherHC.clickCnt
                    thishc.orderCnt += otherHC.orderCnt
                    thishc.payCnt += otherHC.payCnt

                    hcmap.update(cid, thishc)
                }
            }
        }

        override def value: mutable.Map[String, HotCategoryCount] = hcmap
    }
}
